"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 可靠性-数据库系统类
Case Name   : 事务操作产生日志，修改对应xlog文件，再重启数据库
Create At   : 2022/07/07
Owner       : @li-xin12345
Description :
    1. 修改checkpoint_segments、checkpoint_timeout参数
    2. 重启数据库
    3. 执行事务操作
    4. 查询当前xlog文件
    5. 修改当前xlog日志文件
    6. 启动数据库
    7. 再次插入数据
    8. 恢复xlog日志文件
    9. 启动数据库
    10.删除表，恢复参数
Expect      :
    1. 修改参数成功
    2. 重启数据库成功
    3. 执行事务操作成功
    4. 查询当前xlog日志成功
    5. 修改xlog日志文件成功
    6. 启动数据库成功
    7. 再次插入数据成功
    8. 恢复xlog日志成功
    9. 启动数据库成功
    10.删除表成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import RestartDbCluster
from yat.test import Node
from yat.test import macro


class Dbsys(unittest.TestCase):
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.common = Common()
        self.constant = Constant()
        self.table_name = 't_dbsys_case047'
        self.pg_xlogfile_path = ''
        self.bak_xlog_path = ''
        segments_msg = self.primary_sh.executDbSql('show checkpoint_segments;')
        self.logger.info(segments_msg)
        self.segments_default = segments_msg.splitlines()[-2].strip()
        timeout_msg = self.primary_sh.executDbSql('show checkpoint_timeout;')
        self.logger.info(timeout_msg)
        self.timeout_default = timeout_msg.splitlines()[-2].strip()

    def test_dbsys(self):
        text = '-----step1:修改checkpoint_segments、checkpoint_timeout参数;' \
               'expect:修改参数成功-----'
        self.logger.info(text)
        result = self.primary_sh.executeGsguc('reload',
                                              self.constant.GSGUC_SUCCESS_MSG,
                                              'checkpoint_segments=1600')
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show checkpoint_segments;')
        self.logger.info(show_msg)
        self.assertEquals('1600', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)
        result = self.primary_sh.executeGsguc('reload',
                                              self.constant.GSGUC_SUCCESS_MSG,
                                              'checkpoint_timeout=60min')
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show checkpoint_timeout;')
        self.logger.info(show_msg)
        self.assertEquals('1h', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = '-----step2:重启数据库;expect:重启数据库成功-----'
        self.logger.info(text)
        stop_msg = self.primary_sh.stopDbCluster()
        self.logger.info(stop_msg)
        self.assertTrue(stop_msg, '执行失败' + text)
        start_res = self.primary_sh.startDbCluster()
        self.logger.info(start_res)
        self.assertTrue(start_res, '执行失败' + text)

        text = '-----step3:执行事务操作;expect:执行事务操作成功-----'
        self.logger.info(text)
        sql_cmd = f'''set synchronous_commit=on;
            begin 
                drop table if exists {self.table_name};
                create table {self.table_name}(id int,name varchar(20));
                for i in 1..100000 loop
                    insert into {self.table_name} values(i,'aaa');
                end loop;
            end;
            select count(*) from {self.table_name};'''
        self.logger.info(sql_cmd)
        sql_msg = self.primary_sh.executDbSql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn('100000', sql_msg.splitlines()[-2], '执行失败' + text)

        text = '-----step4:查询当前xlog文件;expect:查询当前xlog文件成功-----'
        self.logger.info(text)
        time.sleep(10)
        self.logger.info('-----检查主备是否同步-----')
        node_num = self.common.get_node_num(self.primary_node)
        self.logger.info(f'集群节点数为：{node_num}')
        result = self.primary_sh.check_location_consistency(self.primary_node,
                                                            node_num)
        self.assertTrue(result)
        self.logger.info('-----查询当前xlog文件-----')
        sql_cmd = f"select pg_xlogfile_name(pg_current_xlog_location());"
        self.logger.info(sql_cmd)
        sql_msg = self.primary_sh.executDbSql(sql_cmd)
        self.logger.info(sql_msg)
        pg_xlogfile = sql_msg.splitlines()[-2].strip()
        self.pg_xlogfile_path = os.path.join(macro.PG_XLOG_PATH, pg_xlogfile)
        self.logger.info(self.pg_xlogfile_path)

        text = '-----step5:修改当前xlog日志文件;expect:修改当前xlog日志文件成功-----'
        self.logger.info(text)
        self.logger.info('-----备份xlog文件;expect:备份成功-----')
        self.bak_xlog_path = os.path.join(macro.DB_BACKUP_PATH, pg_xlogfile)
        cp_cmd = f'cp {self.pg_xlogfile_path} {self.bak_xlog_path}'
        self.logger.info(cp_cmd)
        cp_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_msg)
        self.assertEquals('', cp_msg, '执行失败' + text)
        modify_cmd = f"echo '$#@#%GFVJjbffffffd' > {self.pg_xlogfile_path}"
        self.logger.info(modify_cmd)
        modify_msg = self.common.get_sh_result(self.primary_node, modify_cmd)
        self.logger.info(modify_msg)
        self.assertEquals('', modify_msg, '执行失败' + text)

        text = '-----step6:启动数据库;expect:启动数据库成功-----'
        self.logger.info(text)
        stop_msg = self.primary_sh.stopDbCluster()
        self.logger.info(stop_msg)
        self.assertTrue(stop_msg, '停库失败')
        start_res = self.primary_sh.startDbCluster()
        self.logger.info(start_res)
        self.assertTrue(start_res, '启动失败')

        text = '-----step7:再次插入数据;expect:再次插入数据成功-----'
        self.logger.info(text)
        sql_cmd = f'''set synchronous_commit=on;
            begin 
                for i in 1..100000 loop 
                    insert into {self.table_name} values(i,'bbb');
                end loop;
            end;
            select count(*) from {self.table_name};'''
        self.logger.info(sql_cmd)
        sql_msg = self.primary_sh.executDbSql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn('200000', sql_msg.splitlines()[-2], '执行失败' + text)

    def tearDown(self):
        text_8 = '-----step8:恢复xlog日志文件;expect:恢复xlog日志文件成功-----'
        self.logger.info(text_8)
        cp_cmd = f'cp -f {self.bak_xlog_path} {self.pg_xlogfile_path}'
        self.logger.info(cp_cmd)
        cp_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_msg)

        text_9 = '-----step9:启动数据库;expect:启动数据库成功-----'
        self.logger.info(text_9)
        stop_msg = self.primary_sh.stopDbCluster()
        self.logger.info(stop_msg)
        start_res = self.primary_sh.startDbCluster()
        self.logger.info(start_res)

        text_10 = '-----step10:删除表，恢复参数;expect:删除表成功，恢复参数成功-----'
        self.logger.info(text_10)
        drop_cmd = f'drop table {self.table_name};'
        self.logger.info(drop_cmd)
        drop_msg = self.primary_sh.executDbSql(drop_cmd)
        self.logger.info(drop_msg)
        segments_res = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'checkpoint_segments={self.segments_default}')
        show_segments = self.primary_sh.executDbSql('show checkpoint_segments;')
        self.logger.info(show_segments)
        timeout_res = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'checkpoint_timeout={self.timeout_default}')
        show_timeout = self.primary_sh.executDbSql('show checkpoint_timeout;')
        self.logger.info(show_timeout)

        self.logger.info('-----断言tearDown-----')
        self.assertEquals('', cp_msg, '执行失败' + text_8)
        self.assertTrue(stop_msg, '执行失败' + text_9)
        self.assertTrue(start_res, '执行失败' + text_9)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      drop_msg, '执行失败' + text_10)
        self.assertTrue(segments_res, '执行失败' + text_10)
        self.assertEquals(self.segments_default,
                          show_segments.splitlines()[-2].strip(),
                          '执行失败' + text_10)
        self.assertTrue(timeout_res, '执行失败' + text_10)
        self.assertEquals(self.timeout_default,
                          show_timeout.splitlines()[-2].strip(),
                          '执行失败' + text_10)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
